package com.bw.gmall.realtime.dwd.db.split.function;

import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.common.bean.TableProcessDwd;
import com.bw.gmall.realtime.common.util.JdbcUtil;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.util.HashMap;
import java.util.List;

public class DwdProcessFunction extends BroadcastProcessFunction<JSONObject, TableProcessDwd, Tuple2<JSONObject,TableProcessDwd>> {
    private final MapStateDescriptor<String, TableProcessDwd> mapStateDescriptor;
    private HashMap<String, TableProcessDwd> map;

    public DwdProcessFunction(MapStateDescriptor<String, TableProcessDwd> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        map = new HashMap<>();

        Connection connection = JdbcUtil.getMysqlConnection();

        List<TableProcessDwd> tableProcessDwds = JdbcUtil.queryList(connection, "select * from gmall2023_config.table_process_dwd", TableProcessDwd.class, true);

        for (TableProcessDwd tableProcessDwd : tableProcessDwds) {
            map.put(tableProcessDwd.getSourceTable() + "-" + tableProcessDwd.getSourceType(), tableProcessDwd);
        }
    }

    @Override
    public void processElement(JSONObject jsonObject, BroadcastProcessFunction<JSONObject, TableProcessDwd, Tuple2<JSONObject, TableProcessDwd>>.ReadOnlyContext readOnlyContext, Collector<Tuple2<JSONObject, TableProcessDwd>> collector) throws Exception {

        //获取状态 根据主流 来拼接key 值 ，2.数据是事实表 直接过滤掉
        ReadOnlyBroadcastState<String, TableProcessDwd> broadcastState = readOnlyContext.getBroadcastState(mapStateDescriptor);

        String table = jsonObject.getString("table");
        String type = jsonObject.getString("type");
        String key = table + "-" + type;

        TableProcessDwd tableProcessDwd = broadcastState.get(key);
        //是否有数据，如果没有数据，有两种情况，1，广播流 来晚了采用 hashmap取值，
        if (tableProcessDwd == null) {

            //考虑 是因为 广播流 迟到 采用 hashmap
            tableProcessDwd = map.get(key);
        }

        //过滤到需要的数据
        if (tableProcessDwd != null) {

            collector.collect(Tuple2.of(jsonObject, tableProcessDwd));
        }

    }

    @Override
    public void processBroadcastElement(TableProcessDwd tableProcessDwd, BroadcastProcessFunction<JSONObject, TableProcessDwd, Tuple2<JSONObject, TableProcessDwd>>.Context context, Collector<Tuple2<JSONObject, TableProcessDwd>> collector) throws Exception {

        //获取 状态对象
        BroadcastState<String, TableProcessDwd> broadcastState = context.getBroadcastState(mapStateDescriptor);

        String op = tableProcessDwd.getOp();
        String sourceType = tableProcessDwd.getSourceType();


        String key = tableProcessDwd.getSourceTable() + "-" + sourceType;


        if ("d".equals(op)) {

            broadcastState.remove(key);
            map.remove(key);

        } else {

            //以表名 + 类型进行过滤
            broadcastState.put(key, tableProcessDwd);
        }
    }
}
